package biz.ws.controller;

import api.common.result.Result;
import biz.ws.ws.WsCst;
import com.alibaba.nacos.client.naming.NacosNamingService;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaProducerException;
import org.springframework.kafka.core.KafkaSendCallback;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListenerContainer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @创建人 wsdc
 * @时间 2021/4/2
 * @描述 这里是手动测试kafka listen的开关
 */
@RestController
@RequestMapping("/kafka")
@Api
public class KakfaListenerController {
    @Resource
    NacosNamingService nacosNamingService;

    @Resource
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @Resource
    KafkaTemplate kafka;

    @GetMapping("/startOrStop")
    @ApiOperation(value = "启动或者暂停kafka消费者",tags="kafka")
    public Result start(){
        MessageListenerContainer c1 = kafkaListenerEndpointRegistry.getListenerContainer("kafkaMessageReceiveTemp");
        if(!c1.isRunning()){
            c1.start();
        }

        if(c1.isPauseRequested()){
            c1.resume();
            return new Result().data("resume");
        }else{
            c1.pause();
            return new Result().data("pause");
        }
    }

    AtomicInteger ai1 = new AtomicInteger(0);
    @GetMapping("/send")
    @ApiOperation(value = "发送消息",tags="kafka")
    public Result send(@RequestParam(required = false) Integer par){
        if(par == null){
            par = 0;
        }

        kafka.send(WsCst.TopicName.WS_SEND,par, IdWorker.get32UUID(), String.format("{\"a\":%d}",ai1.incrementAndGet()))
                .addCallback(new KafkaSendCallback() {
                    @Override
                    public void onFailure(KafkaProducerException ex) {
                        // TODO: 2021/4/2 异常处理
                    }

                    @Override
                    public void onSuccess(Object result) {
                        // TODO: 2021/4/2 成功处理
                    }
                });

        return new Result();
    }
}
